package com.xix.canal.client.writer.mysql.service;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.xix.canal.client.writer.mysql.config.MysqlConfig;
import com.xix.canal.client.writer.mysql.support.BatchExecutor;
import com.xix.canal.client.writer.mysql.support.SingleDml;
import com.xix.canal.client.writer.mysql.support.SyncUtil;
import com.xixi.canal.client.support.Dml;
import com.xixi.canal.client.support.SQLUtil;
import com.xixi.canal.client.support.ThreadPoolUtils;
import com.xixi.canal.client.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Types;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.function.BiFunction;
import java.util.function.Function;

public class MysqlSyncService {
    private static Logger logger = LoggerFactory.getLogger(MysqlSyncService.class);

    private Map<String, BatchExecutor> executorMap = null;

    private ExecutorService executors = null;

    private Map<String, Map<String, Integer>> columnTypeCache = null;

    public MysqlSyncService(Map<String, DruidDataSource> dataSourceMap) {

        executorMap = new ConcurrentHashMap<>();

        for (Map.Entry<String, DruidDataSource> entry : dataSourceMap.entrySet()) {
            executorMap.putIfAbsent(entry.getKey(), new BatchExecutor(entry.getValue()));
        }
        executors = ThreadPoolUtils.newFixedThreadPool(ThreadPoolUtils.MAX_THREADS,5000L);

        columnTypeCache = new ConcurrentHashMap<>();
    }


    public void close() {


        if (executors != null) {
            executors.shutdown();
        }

        if (executorMap != null && !executorMap.isEmpty()) {
            for (Map.Entry<String, BatchExecutor> entry : executorMap.entrySet()) {
                entry.getValue().close();
            }
            executorMap.clear();
        }


    }

    public void sync(Map<String, Map<String, MysqlConfig>> dbConfigMap, List<Dml> list, Properties properties) {
        sync(list, (dml, dataList) -> {
            String configKey  = StringUtil.getDataSourceKey(dml.getDestination(), dml.getDatabase(), dml.getTable());

            if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtil.hasText(dml.getSql())) {
                columnTypeCache.remove(configKey);
                return false;
            }

            Map<String, MysqlConfig> mysqlConfig = dbConfigMap.get(configKey);

            if (null == mysqlConfig) return false;

            if (mysqlConfig.values().isEmpty()) return false;

            List<SingleDml> singleDmlList = SyncUtil.dml2SingleDmlList(dml);

            for (Map.Entry<String, MysqlConfig> entry : mysqlConfig.entrySet()) {

                MysqlConfig config = entry.getValue();

                String dataSourceKey  = StringUtil.getDataSourceKey(dml.getDestination(),
                        config.getTarget().getSourceDB());

                singleDmlList.forEach(singleDml -> {

                    if (ignoreFieldData(singleDml.getColumnData(), config.getIgnoreField())) {
                        SyncItem syncItem = new SyncItem(config, singleDml);
                        List<SyncItem> syncItems = dataList.computeIfAbsent(dataSourceKey, k -> new ArrayList<>());
                        syncItems.add(syncItem);
                    }
                });
            }
            singleDmlList.clear();
            singleDmlList = null;
            return true;
        });
    }

    private boolean ignoreFieldData(Map<String, Dml.ColumnData> columnData,
                                    List<String> ignoreField) {
        if (ignoreField == null || ignoreField.isEmpty()) return true;

        if (columnData == null || columnData.isEmpty()) return false;

        Set<String> updateColumns = new HashSet<>();
        for (Map.Entry<String, Dml.ColumnData> entry : columnData.entrySet()) {
            Dml.ColumnData value = entry.getValue();
            if (value.getUpdated()) {
                updateColumns.add(entry.getKey());
            }
        }
        boolean ignore = false;
        for (String column : updateColumns) {
            if (!ignoreField.contains(column)) {
                ignore = true;
                break;
            }
        }
        return ignore;
    }

    public void sync(List<Dml> list, BiFunction<Dml, Map<String, List<SyncItem>>, Boolean> function) {
        try {
            boolean execute = false;
            Map<String, List<SyncItem>> dataList = new LinkedHashMap<>();
            for (Dml dml : list) {
                if (!execute) {
                    execute = function.apply(dml, dataList);
                } else {
                    function.apply(dml, dataList);
                }
            }
            if (execute) {
                List<Future<Boolean>> futureList = new ArrayList<>();
                for (Map.Entry<String, List<SyncItem>> entry : dataList.entrySet()) {
                    String key = entry.getKey();
                    List<SyncItem> syncItemList = entry.getValue();
                    BatchExecutor batchExecutor = executorMap.get(key);

                    if (null == batchExecutor) {
                        logger.error("## [{}]Not found data source executor.", key);
                        continue;
                    }
                    futureList.add(executors.submit(() -> {
                        try {
                            syncItemList.forEach(item -> sync(batchExecutor, item.mysqlConfig, item.singleDml));
                            syncItemList.clear();
                            batchExecutor.commit();
                            return true;
                        } catch (SQLException e) {
                            batchExecutor.rollback();
                            throw new RuntimeException(e);
                        }
                    }));
                }

                futureList.forEach(future -> {
                    try {
                        future.get();
                    } catch (InterruptedException  | ExecutionException e) {
                       throw new RuntimeException(e);
                    }
                });
                dataList.clear();
                dataList = null;
            }
        } finally {
            if (null != executorMap && !executorMap.isEmpty()) {
                for (Map.Entry<String, BatchExecutor> entry : executorMap.entrySet()) {
                    entry.getValue().close();
                }
            }
        }
    }

    private void sync(BatchExecutor batchExecutor, MysqlConfig config, SingleDml dml) {
        if (config == null) return;
        try {
            MysqlConfig.TargetConfig target = config.getTarget();
            String resultType = target.getResultType();
            if ("json".equalsIgnoreCase(resultType)) {
                insertJson(batchExecutor, config, dml);
            } else {
                String type = dml.getEventType();
                if (type != null && type.equalsIgnoreCase("INSERT")) {
                    insert(batchExecutor, config, dml);
                } else if (type != null && type.equalsIgnoreCase("UPDATE")) {
                    update(batchExecutor, config, dml);
                } else if (type != null && type.equalsIgnoreCase("DELETE")) {
                    delete(batchExecutor, config, dml);
                } else if (type != null && type.equalsIgnoreCase("TRUNCATE")) {
                    truncate(batchExecutor, config);
                }
            }
            if (logger.isDebugEnabled()) {
                logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }

    private void insertJson(BatchExecutor batchExecutor, MysqlConfig config, SingleDml dml) throws SQLException {
        Map<String, Dml.ColumnData> columnData = dml.getColumnData();
        if (columnData == null || columnData.isEmpty()) {
            return;
        }
        MysqlConfig.TargetConfig target = config.getTarget();

        StringBuilder insertSql = new StringBuilder();
        insertSql.append("INSERT INTO ").append(SyncUtil.getDbTableName(target)).append(" (");
        insertSql.append("destination,source_table,target_table,event_type,ref_id,ref_key,event_data,state,creater,create_time,updater,update_time");
        insertSql.append(") VALUES (?,?,?,?,?,?,?,0,'canal',now(),'canal',now())");

        List<Map<String, ?>> values = new ArrayList<>();
        BatchExecutor.setValue(values, Types.VARCHAR, dml.getDestination());
        BatchExecutor.setValue(values, Types.VARCHAR, target.getSourceTable());
        BatchExecutor.setValue(values, Types.VARCHAR, target.getTargetTable());
        BatchExecutor.setValue(values, Types.VARCHAR, dml.getEventType());


        Map<String, String> pkColumns = target.getPkColumns();

        if (pkColumns != null && !pkColumns.isEmpty()) {
            int index = 0;
            StringBuilder builder = new StringBuilder();
            for (Map.Entry<String, String> entry : pkColumns.entrySet()) {
                if (index == 0) {
                    Dml.ColumnData data = columnData.get(entry.getValue());
                    if (null != data) {
                        BatchExecutor.setValue(values, Types.BIGINT, data.getValue());//ref_id
                    }
                } else {
                    Dml.ColumnData data = columnData.get(entry.getValue());
                    if (null != data) {
                        builder.append(data.getValue());
                        builder.append("-");
                    }
                    break;
                }
                index++;
            }
            int len = builder.length();
            if (len != 0) {
                builder.delete(len -1 , len);
                BatchExecutor.setValue(values, Types.VARCHAR, builder.toString());//ref_key
            }
        }

        Map<String, String> columnsMap = SyncUtil.getColumnsMap(target, columnData);
        Map<String, Object> rowData = new HashMap<>();

        boolean updated = target.getUpdated() == null ? false : target.getUpdated();

        for (Map.Entry<String, String> entry : columnsMap.entrySet()) {
            String targetColumnName = entry.getKey();
            String srcColumnName = entry.getValue();
            if (srcColumnName == null) {
                srcColumnName = StringUtil.cleanColumn(targetColumnName);
            }
            Dml.ColumnData data = columnData.get(srcColumnName);
            if (data == null) {
                throw new RuntimeException("Target source data column: " + srcColumnName + " not matched");
            }

            if (updated && !(data.getUpdated() != null && data.getUpdated())) continue;

            Object value = data.getValue();
            rowData.put(targetColumnName, value);
        }
        if (rowData.isEmpty()) {
            if (logger.isTraceEnabled()) {
                logger.trace("InsertJson into target table, have not change data。");
            }
            return;
        }
        if (pkColumns != null && !pkColumns.isEmpty()) {

            for (Map.Entry<String, String> entry : pkColumns.entrySet()) {
                String targetColumnName = entry.getKey();
                String srcColumnName = entry.getValue();
                if (srcColumnName == null) {
                    srcColumnName = StringUtil.cleanColumn(targetColumnName);
                }
                Dml.ColumnData data = columnData.get(srcColumnName);
                if (data == null) {
                    throw new RuntimeException("Target source data column: " + srcColumnName + " not matched");
                }
                Object value = data.getValue();
                rowData.put(targetColumnName, value);
            }
        }


        BatchExecutor.setValue(values, Types.BLOB, JSON.toJSONString(rowData,
                SerializerFeature.WriteMapNullValue,
                SerializerFeature.WriteDateUseDateFormat));//event_data
        try {
            batchExecutor.execute(insertSql.toString(), values);
        } catch (SQLException e) {
            if (config.getSkipDupException()
                    && (e.getMessage().contains("Duplicate entry") || e.getMessage().startsWith("ORA-00001:"))) {
                // ignore
            } else {
                throw e;
            }
        }
        if (logger.isTraceEnabled()) {
            logger.trace("InsertJson into target table, sql: {}", insertSql);
        }
    }

    private void insert(BatchExecutor batchExecutor, MysqlConfig config, SingleDml dml) throws SQLException {
        Map<String, Dml.ColumnData> columnData = dml.getColumnData();
        if (columnData == null || columnData.isEmpty()) {
            return;
        }

        MysqlConfig.TargetConfig target = config.getTarget();

        Map<String, String> columnsMap = SyncUtil.getColumnsMap(target, columnData);

        StringBuilder insertSql = new StringBuilder();
        insertSql.append("INSERT INTO ").append(SyncUtil.getDbTableName(target)).append(" (");

        columnsMap.forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(","));
        int len = insertSql.length();
        insertSql.delete(len - 1, len).append(") VALUES (");
        int mapLen = columnsMap.size();
        for (int i = 0; i < mapLen; i++) {
            insertSql.append("?,");
        }
        len = insertSql.length();
        insertSql.delete(len - 1, len).append(")");

        String tableKey = StringUtil.getDataSourceKey(dml.getDestination(), dml.getDatabase(), dml.getTable());
        Map<String, Integer> columnTypeMap = getTargetColumnType(batchExecutor.getConn(), config, tableKey);

        List<Map<String, ?>> values = new ArrayList<>();
        for (Map.Entry<String, String> entry : columnsMap.entrySet()) {
            String targetColumnName = entry.getKey();
            String srcColumnName = entry.getValue();
            if (srcColumnName == null) {
                srcColumnName = StringUtil.cleanColumn(targetColumnName);
            }

            Integer type = columnTypeMap.get(StringUtil.cleanColumn(targetColumnName).toLowerCase());
            if (type == null) {
                throw new RuntimeException("Target type column : " + targetColumnName + " not matched");
            }
            Dml.ColumnData data = columnData.get(srcColumnName);
            if (data == null) {
                throw new RuntimeException("Target source data column: " + srcColumnName + " not matched");
            }
            Object value = data.getValue();
            BatchExecutor.setValue(values, type, value);
        }

        try {
            batchExecutor.execute(insertSql.toString(), values);
        } catch (SQLException e) {
            if (config.getSkipDupException()
                    && (e.getMessage().contains("Duplicate entry") || e.getMessage().startsWith("ORA-00001:"))) {
                // ignore
            } else {
                throw e;
            }
        }
        if (logger.isTraceEnabled()) {
            logger.trace("Insert into target table, sql: {}", insertSql);
        }
    }
    private void update(BatchExecutor batchExecutor, MysqlConfig config, SingleDml dml) throws SQLException {
        Map<String, Dml.ColumnData> columnData = dml.getColumnData();
        if (columnData == null || columnData.isEmpty()) {
            return;
        }

        MysqlConfig.TargetConfig target = config.getTarget();

        Map<String, String> columnsMap = SyncUtil.getColumnsMap(target, columnData);

        String tableKey = StringUtil.getDataSourceKey(dml.getDestination(), dml.getDatabase(), dml.getTable());
        Map<String, Integer> columnTypeMap = getTargetColumnType(batchExecutor.getConn(), config, tableKey);

        StringBuilder updateSql = new StringBuilder();
        updateSql.append("UPDATE ").append(SyncUtil.getDbTableName(target)).append(" SET ");
        List<Map<String, ?>> values = new ArrayList<>();
        boolean hasMatched = false;

        boolean updated = target.getUpdated() == null ? false : target.getUpdated();

        for (Map.Entry<String, String> entry : columnsMap.entrySet()) {
            String targetColumnName = entry.getKey();
            String srcColumnName = entry.getValue();
            if (srcColumnName == null) {
                srcColumnName = StringUtil.cleanColumn(targetColumnName);
            }

            Dml.ColumnData data = columnData.get(srcColumnName);

            if (data == null) {
                throw new RuntimeException("Target source data column: " + srcColumnName + " not matched");
            }

            if (updated && !(data.getUpdated() != null && data.getUpdated())) continue;

            updateSql.append(targetColumnName).append("=?, ");
            Integer type = columnTypeMap.get(StringUtil.cleanColumn(targetColumnName).toLowerCase());
            if (type == null) {
                throw new RuntimeException("Target type column : " + targetColumnName + " not matched");
            }
            BatchExecutor.setValue(values, type, data.getValue());
            hasMatched = true;
        }
        if (!hasMatched) {
            logger.warn("Did not matched any columns to update ");
            return;
        }
        int len = updateSql.length();
        updateSql.delete(len - 2, len).append(" WHERE ");

        // 拼接主键
        appendCondition(config.getTarget(), updateSql, columnTypeMap, values, columnTypeMap, dml.getColumnData());
        batchExecutor.execute(updateSql.toString(), values);
        if (logger.isTraceEnabled()) {
            logger.trace("Update target table, sql: {}", updateSql);
        }
    }



    private void delete(BatchExecutor batchExecutor, MysqlConfig config, SingleDml dml) throws SQLException {
        Map<String, Dml.ColumnData> columnData = dml.getColumnData();
        if (columnData == null || columnData.isEmpty()) {
            return;
        }
        MysqlConfig.TargetConfig target = config.getTarget();

        String tableKey = StringUtil.getDataSourceKey(dml.getDestination(), dml.getDatabase(), dml.getTable());

        Map<String, Integer> columnTypeMap = getTargetColumnType(batchExecutor.getConn(), config, tableKey);

        StringBuilder sql = new StringBuilder();
        sql.append("DELETE FROM ").append(SyncUtil.getDbTableName(target)).append(" WHERE ");

        List<Map<String, ?>> values = new ArrayList<>();
        // 拼接主键
        appendCondition(config.getTarget(), sql, columnTypeMap, values, columnTypeMap, dml.getColumnData());
        batchExecutor.execute(sql.toString(), values);

        if (logger.isTraceEnabled()) {
            logger.trace("Delete from target table, sql: {}", sql);
        }
    }

    private void appendCondition(MysqlConfig.TargetConfig targetConfig,
                                 StringBuilder sql,
                                 Map<String, Integer> columnTypeMap,
                                 List<Map<String, ?>> values,
                                 Map<String, Integer> typeMap,
                                 Map<String, Dml.ColumnData> columnDataMap) {
        for (Map.Entry<String, String> entry : targetConfig.getPkColumns().entrySet()) {
            String targetColumn = entry.getKey();
            String sourceColumn = entry.getValue();
            if (sourceColumn == null) {
                sourceColumn = StringUtil.cleanColumn(targetColumn);
            }
            sql.append(targetColumn).append("=? AND");

            Integer type = typeMap.get(StringUtil.cleanColumn(targetColumn).toLowerCase());
            if (type == null) {
                throw new RuntimeException("Target type column : " + targetColumn + " not matched");
            }

            Dml.ColumnData columnData = columnDataMap.get(sourceColumn);
            if (columnData == null) {
                throw new RuntimeException("Target source data column: " + sourceColumn + " not matched");
            }
            BatchExecutor.setValue(values, type, columnData.getValue());

        }
        int len = sql.length();
        sql.delete(len - 4, len);
    }

    private void truncate(BatchExecutor batchExecutor, MysqlConfig config) throws SQLException {
        MysqlConfig.TargetConfig target = config.getTarget();
        StringBuilder sql = new StringBuilder();
        sql.append("TRUNCATE TABLE ").append(SyncUtil.getDbTableName(target));
        batchExecutor.execute(sql.toString(), new ArrayList<>());
        if (logger.isTraceEnabled()) {
            logger.trace("Truncate target table, sql: {}", sql);
        }
    }

    public Map<String, Integer> getTargetColumnType(Connection conn, MysqlConfig mysqlConfig, String tableKey) {

        Map<String, Integer> typeMap = columnTypeCache.get(tableKey);
        if (typeMap == null) {
            synchronized (MysqlSyncService.class) {
                typeMap = columnTypeCache.get(tableKey);
                if (typeMap == null) {
                    typeMap = new LinkedHashMap<>();
                    String sql = "SELECT * FROM " + SyncUtil.getDbTableName(mysqlConfig.getTarget()) + " WHERE 1=2";
                    Map<String, Integer> finalTypeMap = typeMap;
                    SQLUtil.sqlRS(conn, sql, resultSet -> {
                        try {
                            ResultSetMetaData metaData = resultSet.getMetaData();
                            int columnCount = metaData.getColumnCount();
                            for (int i = 1; i <= columnCount; i++) {
                                finalTypeMap.put(metaData.getColumnName(i).toLowerCase(),
                                        metaData.getColumnType(i));
                            }
                            columnTypeCache.put(tableKey, finalTypeMap);
                        } catch (SQLException e) {
                            logger.error(e.getMessage(), e);
                        }
                    });
                }
            }
        }
        return typeMap;
    }

    public static class SyncItem {
        private MysqlConfig mysqlConfig;
        private SingleDml singleDml;

        public SyncItem(MysqlConfig mysqlConfig, SingleDml singleDml) {
            this.mysqlConfig = mysqlConfig;
            this.singleDml = singleDml;
        }

        public MysqlConfig getMysqlConfig() {
            return mysqlConfig;
        }

        public void setMysqlConfig(MysqlConfig mysqlConfig) {
            this.mysqlConfig = mysqlConfig;
        }

        public SingleDml getSingleDml() {
            return singleDml;
        }

        public void setSingleDml(SingleDml singleDml) {
            this.singleDml = singleDml;
        }
    }


}
